package cn.naughtior.shch.utils;


import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.client.impl.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * ${DESCRIPTION}
 *
 * @author shitou<shitou@in66.com>
 * @create 2017-01-14 17:59
 */

public class JiuyanChannel extends ChannelN {

    /**
     * Construct a new channel on the given connection with the given
     * channel number. Usually not called directly - call
     * Connection.createChannel instead.
     *
     * @param connection    The connection associated with this channel
     * @param channelNumber The channel number to be associated with this channel
     * @param workService   service for managing this channel's consumer callbacks
     * @see
     */
    public JiuyanChannel(AMQConnection connection, int channelNumber, ConsumerWorkService workService) {
        super(connection, channelNumber, workService);
    }

    /** Public API - {@inheritDoc} */
    @Override
    public void close()
            throws IOException
    {
        close(AMQP.REPLY_SUCCESS, "OK");
    }

    /** Public API - {@inheritDoc} */
    @Override
    public void close(int closeCode, String closeMessage)
            throws IOException
    {
        close(closeCode, closeMessage, true, null, false);
    }

    /** Public API - {@inheritDoc} */
    @Override
    public void abort()
            throws IOException
    {
        abort(AMQP.REPLY_SUCCESS, "OK");
    }

    /** Public API - {@inheritDoc} */
    @Override
    public void abort(int closeCode, String closeMessage)
            throws IOException
    {
        close(closeCode, closeMessage, true, null, true);
    }

    // TODO: method should be private
    /**
     * Protected API - Close channel with code and message, indicating
     * the source of the closure and a causing exception (null if
     * none).
     * @param closeCode the close code (See under "Reply Codes" in the AMQP specification)
     * @param closeMessage a message indicating the reason for closing the connection
     * @param initiatedByApplication true if this comes from an API call, false otherwise
     * @param cause exception triggering close
     * @param abort true if we should close and ignore errors
     * @throws IOException if an error is encountered
     */
    @Override
    public void close(int closeCode,
                      String closeMessage,
                      boolean initiatedByApplication,
                      Throwable cause,
                      boolean abort)
            throws IOException
    {
        // First, notify all our dependents that we are shutting down.
        // This clears isOpen(), so no further work from the
        // application side will be accepted, and any inbound commands
        // will be discarded (unless they're channel.close-oks).
        AMQImpl.Channel.Close reason = new AMQImpl.Channel.Close(closeCode, closeMessage, 0, 0);
        ShutdownSignalException signal = new ShutdownSignalException(false,
                initiatedByApplication,
                reason,
                this);
        if (cause != null) {
            signal.initCause(cause);
        }

        BlockingRpcContinuation<AMQCommand> k = new BlockingRpcContinuation<AMQCommand>(){
            @Override
            public AMQCommand transformReply(AMQCommand command) {
               //finishProcessShutdownSignal();
                return command;
            }};
        boolean notify = false;
        try {
            // Synchronize the block below to avoid race conditions in case
            // connnection wants to send Connection-CloseOK
            synchronized (_channelMutex) {
                startProcessShutdownSignal(signal, !initiatedByApplication, true);
                quiescingRpc(reason, k);
            }

            // Now that we're in quiescing state, channel.close was sent and
            // we wait for the reply. We ignore the result.
            // (It's NOT always close-ok.)
            notify = true;
            k.getReply(-1);
        } catch (TimeoutException ise) {
            // Will never happen since we wait infinitely
        } catch (ShutdownSignalException sse) {
            if (!abort)
                throw sse;
        } catch (IOException ioe) {
            if (!abort)
                throw ioe;
        } finally {
            if (abort || notify) {
                // Now we know everything's been cleaned up and there should
                // be no more surprises arriving on the wire. Release the
                // channel number, and dissociate this ChannelN instance from
                // our connection so that any further frames inbound on this
                // channel can be caught as the errors they are.
                releaseChannel();
                notifyListeners();
            }
        }
    }

    private void startProcessShutdownSignal(ShutdownSignalException signal,
                                            boolean ignoreClosed,
                                            boolean notifyRpc)
    {   super.processShutdownSignal(signal, ignoreClosed, notifyRpc);
    }

    private void releaseChannel() {
        getConnection().disconnectChannel(this);
    }

}
